package com.om.authing.event;

import Utils.HttpClientUtils;
import com.om.authing.model.EventLog;
import com.om.common.CommonObject;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.broadcast.Broadcast;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;

import java.util.Iterator;
import java.util.Map;

public class AppEventMain {
    public static void main(String[] args) throws ClassNotFoundException, IllegalAccessException, InstantiationException {
        String taskCount = args[2];
        String inputPath = args[3];
        String esIndexName = args[4];
        String className = args[5];

        AppEvent appEvent = (AppEvent) Class.forName(className).newInstance();
        appEvent.setEsIndexName(esIndexName);

        SparkConf sparkConf = appEvent.getSparkConf(args);
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);

        Broadcast<AppEvent> AppEventBc = jsc.broadcast(appEvent);
        JavaRDD<String> srcRdd = jsc.textFile(inputPath, Integer.parseInt(taskCount));
        parseEvent(srcRdd, AppEventBc);
    }

    private static void parseEvent(JavaRDD<String> srcRdd, Broadcast<AppEvent> AppEventBc) {
        srcRdd.foreachPartition((VoidFunction<Iterator<String>>) it -> {
            RestHighLevelClient restHighLevelClient = HttpClientUtils.restClient();
            BulkRequest request = new BulkRequest();

            AppEvent appEvent = AppEventBc.value();
            while (it.hasNext()) {
                String data = it.next();
                EventLog eventLog = appEvent.checkLog(data);
                if (eventLog == null) continue;
                eventLog = appEvent.parseBody(eventLog);
                Map source = CommonObject.objectMapper.convertValue(eventLog, Map.class);
                request.add(new IndexRequest(appEvent.getEsIndexName(), "_doc", DigestUtils.md5Hex(data)).source(source));
            }
            if (request.requests().size() != 0) restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
            restHighLevelClient.close();
        });
        System.out.println("*** Event end ***");
    }
}
